package demo.db.elasticSearch.highClient;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import demo.db.elasticSearch.UserMsgVO;
import demo.db.redis.vo.Slowlog;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

/**
 * <h1>Java High Level REST Client</h1>从6.0.0开始加入的，目的是以java面向对象的方式来进行请求、响应处理。
 * 每个API 支持 同步/异步 两种方式，同步方法直接返回一个结果对象。异步的方法以async为后缀，通过listener参数来通知结果。
 * 
 * 兼容性说明：依赖 java1.8 和 Elasticsearch core project,请使用与服务端ES版本一致的客户端版本
 * 
 * @author hanjy
 *
 */
public class DocumentDemo {

    private static Logger logger = LoggerFactory.getLogger(DocumentDemo.class);

    @Test
    public void updateDemo1() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("mess", "_doc", "3");

            Map<String, Object> parameters = new HashMap<>();
            parameters.put("count", 4);

            Script inline = new Script(ScriptType.INLINE, "painless", "ctx._source.field += params.count", parameters);
            request.script(inline);
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void updateDemo2() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("mess", "_doc", "3");

            Map<String, Object> parameters = new HashMap<>();
            parameters.put("count", 4);

            Script stored = new Script(ScriptType.STORED, null, "increment-field", parameters);
            request.script(stored);
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void updateDemo3() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("mess", "_doc", "3");
            UserMsgVO vo = new UserMsgVO();
            vo.setUser("Stan");
            vo.setMessage("hello elastic search!!!");
            String jsonString = JSONObject.toJSONString(vo);
            request.doc(jsonString, XContentType.JSON);
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void updateDemo4() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("mess", "_doc", "3");
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("updated", new Date());
            jsonMap.put("reason", "daily update");
            request.doc(jsonMap);
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void updateDemo5() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("mess", "_doc", "3");
            XContentBuilder builder = XContentFactory.jsonBuilder();
            builder.startObject();
            {
                builder.timeField("updated", new Date());
                builder.field("reason", "daily update");
            }
            builder.endObject();
            request.doc(builder);
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void updateDemo6() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("posts", "doc", "1");
            request.doc("updated", new Date(), "reason", "daily update");
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * If the document does not already exist, it is possible to define some content
     * that will be inserted as a new document using the upsert method:
     */
    @Test
    public void upsertDemo1() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            // 1、创建获取文档请求
            UpdateRequest request = new UpdateRequest("posts", "doc", "1");
            UserMsgVO vo = new UserMsgVO();
            vo.setUser("Stan");
            vo.setMessage("hello elastic search!!!");
            String jsonString = JSONObject.toJSONString(vo);
            request.upsert(jsonString, XContentType.JSON);
            request.upsert("updated", new Date(), "reason", "daily update");
            // 重试次数
            request.retryOnConflict(3);
            UpdateResponse updateResponse = null;
            try {
                updateResponse = client.update(request, RequestOptions.DEFAULT);
            } catch (ElasticsearchException e) {
                if (e.status() == RestStatus.NOT_FOUND) {
                    logger.warn("When a UpdateRequest is performed against a document that does not exist, "
                            + "the response has 404 status code,"
                            + " an ElasticsearchException gets thrown which needs to be handled ");
                }
                if (e.status() == RestStatus.CONFLICT) {
                    logger.warn("If there is a version conflict, an ElasticsearchException will be thrown:");
                }
            }
            String index = updateResponse.getIndex();
            String type = updateResponse.getType();
            String id = updateResponse.getId();
            long version = updateResponse.getVersion();
            if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {

            } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {

            } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {

            } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {

            }
            GetResult result = updateResponse.getGetResult();
            if (result.isExists()) {
                String sourceAsString = result.sourceAsString();
                Map<String, Object> sourceAsMap = result.sourceAsMap();
                byte[] sourceAsBytes = result.source();
            } else {

            }
            // It is also possible to check for shard failures:
            ReplicationResponse.ShardInfo shardInfo = updateResponse.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {

            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                }
            }
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 版本冲突
     */
    @Test
    public void demoVersionConflict() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            IndexRequest request = new IndexRequest("posts", "doc", "1").source("field", "value").version(1);
            try {
                IndexResponse response = client.index(request, RequestOptions.DEFAULT);
                EsUtils.printIndexResponse(response);
            } catch (ElasticsearchException e) {
                if (e.status() == RestStatus.CONFLICT) {

                }
            }

        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 主键冲突
     */
    @Test
    public void demoIdConflict() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            IndexRequest request = new IndexRequest("posts", "doc", "1").source("field", "value")
                    .opType(DocWriteRequest.OpType.CREATE);
            try {
                IndexResponse response = client.index(request, RequestOptions.DEFAULT);
                EsUtils.printIndexResponse(response);
            } catch (ElasticsearchException e) {
                if (e.status() == RestStatus.CONFLICT) {
                    logger.warn(
                            "Same will happen in case opType was set to create and a document with same index, type and id already existed");
                }
            }

        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 新增一条记录
     * 
     * @param client
     * @param index
     * @param type
     * @param documentId
     * @param documentSource
     * @throws IOException
     */
    public static void indexRequest(RestHighLevelClient client, String index, String type, String documentId,
            JSONObject documentSource) throws IOException {
        IndexRequest request = new IndexRequest(index, type, documentId);
        logger.info("source={}", documentSource);
        request.source(documentSource.toString(), XContentType.JSON);
        IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
        // client.indexAsync(request, RequestOptions.DEFAULT, defaultActionListener);
    }

    /**
     * 新增一条记录
     * 
     * @param client
     * @param index
     * @param type
     * @param documentId
     * @param documentSource
     * @throws IOException
     */
    public static void indexRequest(RestHighLevelClient client, String index, String type, String documentId,
            Map<String, Object> documentSource) throws IOException {
        IndexRequest request = new IndexRequest(index, type, documentId);
        logger.info("source={}", documentSource);
        // Document source provided as a Map which gets automatically converted to JSON
        // format
        request.source(documentSource);
        IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
        // client.indexAsync(request, RequestOptions.DEFAULT, defaultActionListener);
        EsUtils.printIndexResponse(indexResponse);
    }

    /**
     * 新增一条记录
     * 
     * @param client
     * @param index
     * @param type
     * @param documentId
     * @param documentSource
     * @throws IOException
     */
    public static void indexRequest(RestHighLevelClient client, String index, String type, String documentId,
            Object... documentSource) throws IOException {
        IndexRequest request = new IndexRequest(index, type, documentId);
        logger.info("source={}", documentSource);
        // Document source provided as a Map which gets automatically converted to JSON
        // format
        request.source(documentSource);
        IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
        // client.indexAsync(request, RequestOptions.DEFAULT, defaultActionListener);
        EsUtils.printIndexResponse(indexResponse);
    }

    /**
     * 新增一条记录
     * 
     * @param client
     * @param index
     * @param type
     * @param documentId
     * @param documentSource
     * @throws IOException
     */
    public static void indexRequest2(RestHighLevelClient client, String index, String type, String documentId,
            Map<String, Object> documentSource) throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        for (Entry<String, Object> entry : documentSource.entrySet()) {
            builder.field(entry.getKey(), entry.getValue());
        }
        builder.endObject();
        IndexRequest request = new IndexRequest(index, type, documentId);
        logger.info("source={}", builder);
        // Document source provided as an XContentBuilder object,
        // the Elasticsearch built-in helpers to generate JSON content
        request.source(builder);
        IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
        EsUtils.printIndexResponse(indexResponse);
        // client.indexAsync(request, RequestOptions.DEFAULT, defaultActionListener);
    }

    /**
     * 索引文档，即往索引里面放入文档数据.类似于数据库里面向表里面插入一行数据，一行数据就是一个文档
     * 
     * @param args
     */
    @Test
    public void testAddDocument1() {
        String index = "slow_redis_index.2021-04-10";
        Slowlog vo = new Slowlog("192.168.86.36", 6380, 0, System.currentTimeMillis(), 10.0, "get a");
        IndexRequest indexRequest = new IndexRequest(index);
        JSONObject jsonObject = (JSONObject) JSON.toJSON(vo);
        List<Object> list = new ArrayList<>();
        for (Entry<String, Object> entry : jsonObject.entrySet()) {
            list.add(entry.getKey());
            list.add(entry.getValue());
        }
        indexRequest.source(XContentType.JSON, list.toArray());
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
            System.err.println(indexResponse);
            RestStatus status = indexResponse.status();
            System.err.println(status);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    @Test
    public void testAddDocument2() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject().field("redisIp", "192.168.86.35")
                .field("redisPort", 6382).field("id", 0).field("timeStamp", System.currentTimeMillis())
                .field("executionTime", 10.0).field("args", "get a").endObject();

        String index = "slow_redis_index.2021-04-10";
        IndexRequest indexRequest = new IndexRequest(index);
        indexRequest.source(builder);
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
            System.err.println(indexResponse);
            RestStatus status = indexResponse.status();
            System.err.println(status);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 索引文档，即往索引里面放入文档数据.类似于数据库里面向表里面插入一行数据，一行数据就是一个文档
     * 
     */
    @Test
    public void addDocument2Index2() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("user", "kimchy");
            jsonMap.put("postDate", new Date());
            jsonMap.put("message", "trying out Elasticsearch");

            indexRequest(client, "mess", "_doc", null, jsonMap);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 索引文档，即往索引里面放入文档数据.类似于数据库里面向表里面插入一行数据，一行数据就是一个文档
     * 
     * @param args
     */
    @Test
    public void addDocument2Index3() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("user", "kimchy");
            jsonMap.put("postDate", new Date());
            jsonMap.put("message", "trying out Elasticsearch");
            indexRequest2(client, "mess", "_doc", "5", jsonMap);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    /**
     * 索引文档，即往索引里面放入文档数据.类似于数据库里面向表里面插入一行数据，一行数据就是一个文档
     * 
     * @param args
     */
    @Test
    public void addDocument2Index4() {
        try (RestHighLevelClient client = RestHighLevelClientDemo.getClient()) {
            indexRequest(client, "mess", "_doc", "6", "user", "kimchy", "postDate", new Date(), "message",
                    "trying out Elasticsearch");
        } catch (IOException e) {
            logger.error("", e);
        }
    }

}
